// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jing

#include "coin2/exchange/bithumb/feed_v2/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bithumb/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bithumb::feed_v2::impl {

void BithumbFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  static constexpr std::string_view wallet_url("assetsstatus/ALL");
  static constexpr std::string_view kline_url("candlestick/");
  static constexpr std::string_view book_url("orderbook/");
  std::string url_type;
  if (document.HasMember("url")) {
    const std::string_view url = document["url"].GetString();
    if (url.rfind(wallet_url) != std::string::npos) {
      url_type = "wallet";
      if (!CheckChannelFilter(url_type)) {
        return;
      }
    } else if (url.find(kline_url) != std::string::npos) {
      url_type = "kline";
    } else if (url.find(book_url) != std::string::npos) {
      url_type = "orderbook";
    }

    if(strcmp(document["data"]["status"].GetString(),"0000") != 0){
      url_type = "exception";
    }
  }

  if (document.HasMember("type")) {
    const std::string& type = document["type"].GetString();

    if (!CheckChannelFilter(type)) {
      return;
    }

    if (type == "transaction") {
      const auto& content_list = document["content"]["list"];
      for (const auto& transaction : content_list.GetArray()) {
        const auto& native_symbol = transaction["symbol"].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          continue;
        }
        Trade* trade = upd_mgr_.GetTradeMutable();
        BithumbFeedUpdateParser::ParseTrade(transaction, trade, packet.timestamp);
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (type == "orderbookdepth") {
      const auto& content_list = document["content"]["list"];
      std::string native_symbol;
      for (const auto& ask_bid : content_list.GetArray()) {
        if (native_symbol.empty()) {
          native_symbol = ask_bid["symbol"].GetString();
        } else if (native_symbol != ask_bid["symbol"].GetString()) {
          THROW() << "Different symbols in orderbookdepth";
        }
      }
      if (native_symbol.empty()) {
        THROW() << "Empty content list";
      }
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[native_symbol];
        if (stat.snapshot_received || packet.timestamp < SNAPSHOT_ADDED_TS) {
          BithumbFeedUpdateParser::ParseBook(book, content_list, packet.timestamp);
          upd_mgr_.PublishBook();
          stat.update_received = true;
        }
      }
    } else {
      LOG(INFO) << "[Bithumb Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload;
    }
  } else if (url_type == "kline") {
    std::string url = document["url"].GetString();
    if (url.find("/candlestick/") != std::string::npos) {
      // kline
      auto splited_url = Split(url, "/");
      size_t len = splited_url.size();
      CHECK_GE(len, 5);
      const std::string& interval_str = splited_url[len - 1];
      const std::string& native_symbol = splited_url[len - 2];
      std::string channel_str = "candlestick/{}/" + interval_str;
      if (!CheckChannelFilter(channel_str)) {
        return;
      }

      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          return;
      }

      auto interval = BithumbFeedUpdateParser::ParseKlineInterval(interval_str);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        // bithumb kline api doesn't specifiy data time duruation, so all data are snapshot
        if (klines->Size() == 0) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }

        const auto& items = document["data"]["data"].GetArray();
        bool to_publish = false;
        for (const auto& kline_data : items) {
          bool freezed = BithumbFeedUpdateParser::Parse1Kline(
              kline_data, klines, packet.timestamp, interval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else if (url_type == "wallet") { 
    if (document["data"]["data"].IsObject()) {
      for (const auto& [key, value] : document["data"]["data"].GetObject()) {
        std::string_view currency = key.GetString();
        for (const auto& product : upd_mgr_.products()) {
          if (product->base().native_currency() == currency) {
            Wallet* wallet = upd_mgr_.GetWalletMutable();
            BithumbFeedUpdateParser::ParseWallet(value, wallet);
            upd_mgr_.PublishWallet(product->feed_native_symbol());
          }
        }
      }
    }
  } else if (url_type == "orderbook") { 
    if (document["data"]["data"].IsObject()){
      auto content_list = document["data"]["data"].GetObject();
      int64_t ts = stoll(content_list["timestamp"].GetString()) * 1'000'000;
      std::string_view payment_currency = content_list["payment_currency"].GetString();
      for (const auto& currency_data : content_list) {
        if (!currency_data.value.IsObject()) {
          continue;
        }

        std::string_view order_currency = currency_data.value["order_currency"].GetString();
        std::stringstream ss;
        ss << order_currency << "_" << payment_currency;
        std::string native_symbol = ss.str();
        auto* book = upd_mgr_.GetBookMutable(native_symbol);
        if (book != nullptr) {
          auto& stat = topic_stat_[native_symbol];
          if (!stat.snapshot_received || !stat.update_received) {
            BithumbFeedUpdateParser::ParseSnapshot(book, currency_data.value, ts);
            upd_mgr_.PublishBook();
            stat.snapshot_received = true;
          }
        }
      }
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Bithumb Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::bithumb::feed_v2::impl
}  // namespace coin2::exchange::bithumb::feed_v2::impl
